package com.happy.xiao.common.rabbitmq;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.core.env.Environment;

public class BaseMessageContainer {

    @Autowired
    protected CachingConnectionFactory connectionFactory;

    @Autowired
    protected SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Autowired
    protected Environment env;



    /**
     * 构建公共的信息容器
     *
     * @param messageListener
     * @param myQueue
     * @return
     */
    protected SimpleMessageListenerContainer createContainer(Object messageListener, Queue myQueue) {

        // 默认多少消费者
        int concurrentConsumers = env.getProperty("spring.rabbitmq.listener.concurrency", Integer.class);
        // 最大消费者
        int maxConcurrentConsumers = env.getProperty("spring.rabbitmq.listener.max-concurrency", Integer.class);
        // 预拉取信息多少条
        int prefetchCount = env.getProperty("spring.rabbitmq.listener.prefetch", Integer.class);

        return createContainer(messageListener, myQueue, concurrentConsumers, maxConcurrentConsumers, prefetchCount);
    }

    /**
     * 构建公共的信息容器
     *
     * @param messageListener        监听队列的回调
     * @param myQueue                监听队列
     * @param concurrentConsumers    默认消费者数目
     * @param maxConcurrentConsumers 最大消费者数目
     * @param prefetchCount          每个消费者最大拉取消息数量【这个很重要，不设置的话，如果消息服务宕机之后，重启服务时候，
     *                               会不限制消息数量拉取，这样服务器的压力剧增，容易造成服务器宕机】
     * @return
     */
    protected SimpleMessageListenerContainer createContainer(Object messageListener,
                                                           Queue myQueue,
                                                           int concurrentConsumers,
                                                           int maxConcurrentConsumers,
                                                           int prefetchCount) {
        SimpleMessageListenerContainer simple = new SimpleMessageListenerContainer();
        simple.setConnectionFactory(connectionFactory);
        simple.setMessageConverter(new Jackson2JsonMessageConverter());
        // 并发针对消费Listener来说的

        // 默认多少消费者
        if (concurrentConsumers != 0) simple.setConcurrentConsumers(concurrentConsumers);
        // 最大消费者
        if (maxConcurrentConsumers != 0) simple.setMaxConcurrentConsumers(maxConcurrentConsumers);
        // 预拉取信息多少条
        if (prefetchCount != 0) simple.setPrefetchCount(prefetchCount);

        // 消息确认-手动确认机制
        simple.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        simple.setQueues(myQueue);

        if (messageListener != null) simple.setMessageListener(messageListener);
        return simple;
    }
}
